package com.study;

import com.rabbitmq.client.*;


import java.net.URI;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;

public class PublishConfirmsProducer2  {
    public static void main(String[] args) throws Exception {
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setUri(URI.create("amqp://root:123456@192.168.181.131:5672/%2f"));

        Connection connection = connectionFactory.newConnection();
        Channel channel = connection.createChannel();
        //声明发送方确认机制
        AMQP.Confirm.SelectOk selectOk = channel.confirmSelect();

        channel.queueDeclare("queue.confirm",false,false,false,null);
        channel.exchangeDeclare("exchange.confirm", "direct",false,false,null);

        channel.queueBind("queue.confirm","exchange.confirm","confirmroutingkey.key");


        String msg  = "hello world+";


       /* ConfirmCallback confirmCallback = new ConfirmCallback() {
            @Override
            public void handle(long deliveryTage, boolean multiple) throws IOException {
                if(multiple){
                    System.out.println("编号小于等于"+deliveryTage+"已经被确认了");
                }else{
                    System.out.println("编号为"+deliveryTage+"的消息已经被确认了");
                }
            }
        };*/
        ConcurrentNavigableMap<Long,String> outstandingConfirms = new ConcurrentSkipListMap();
        ConfirmCallback clearOutstandingConfirms = (deliveryTage,multiple)->{
            if(multiple){
                System.out.println("编号小于等于"+deliveryTage+"已经被确认了");

                ConcurrentNavigableMap<Long, String> headMap
                        = outstandingConfirms.headMap(deliveryTage, true);
                //清空outstandingConfirms中已经被确认了的消息
                headMap.clear();
            }else{
                //移除已经被确认了的消息
                outstandingConfirms.remove(deliveryTage);
                System.out.println("编号为"+deliveryTage+"的消息已经被确认了");

            }
        };

        //确认channel的监听器，处理确认的消息和不确认的消息
        channel.addConfirmListener(clearOutstandingConfirms,(deliveryTage,multiple)->{
            if(multiple){
                //对于没有被确认的消息可以放入到一个map中，后续对其进行处理
                System.out.println("消息编号小于等于"+deliveryTage+"的消息不被确认");
            }else{
                System.out.println("消息编号为"+deliveryTage+"的消息不被确认");
            }
        });

        //发送消息
        for (int i = 0; i < 1000; i++) {
            //获取下一条即将发送的消息的消息ID
            long nextPublishSeqNo = channel.getNextPublishSeqNo();
            channel.basicPublish("exchange.confirm","confirmroutingkey.key",null,(msg+i).getBytes());

            System.out.println("编号为："+nextPublishSeqNo+"的消息已经发送成功，尚未确认");
            outstandingConfirms.put(nextPublishSeqNo,(msg+i));
        }

        Thread.sleep(2000);

        channel.close();
        connection.close();
    }

}
